package day19;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.Iterator;


public class TransformationTestJava {

    public static void main(String[] args) throws Exception {
        // 获得flink唤醒
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 1.需求1：将元祖内数据求和。map
        //DataSource是DataSet的子类，这里直接改为成DataSet了
        DataSet<Tuple2<Integer, Integer>> pairs = env.fromElements(new Tuple2<Integer, Integer>(1, 2), new Tuple2<Integer, Integer>(3, 4));
        DataSet<Integer> sumed = pairs.map(new IntegerOperations());
        sumed.print();  //输出 3 7

        // 2.需求2：切分字符串。flatMap
        //DataSource是DataSet的子类，这里直接改为成DataSet了
        DataSet<String> text = env.fromElements(
                "hello zhangsan",
                "hello lisi",
                "hello wangwu");
        //这里的返回值DataSet<String> words 最好自己写，因为自动返回的有问题。
        DataSet<String> words = text.flatMap(new FlatMapFunction<String, String>() {
            // 这里flatMap每次传递过来的是一行数据s，我们用一个Collector来保存每一个切分后的单词
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] splited = s.split(" ");
                for (String word : splited)
                    collector.collect(word);
            }
        });
        words.print();

        // 3.需求3：求每个分区的元素个数
        DataSet<Long> counts = text.mapPartition(new MapPartitionFunction<String, Long>() {
            @Override
            public void mapPartition(Iterable<String> iterable, Collector<Long> collector) throws Exception {
//                 这iterable参数是代表某个分区的数据组成的迭代器。
//                统计每行数据的单词个数:如：2 2 2？？
//                Iterator<String> it = iterable.iterator();
//                while (it.hasNext()){
//                    String str = it.next();
//                    String[] splitedStr = str.split(" ");
//                    collector.collect((long) splitedStr.length);
//                }

                long i = 0;
                Iterator<String> it = iterable.iterator();
                while (it.hasNext()){
                    it.next();
                    i += 1;
                }
                collector.collect(i);
            }
        });
        counts.print();

        //4.需求4：filter：过滤大于3的数。（filter就是返回满足条件的数据。将不满足条件的数据给过滤掉）
        DataSet<Integer> nums = env.fromElements(1, 2, 3, 4, 5, 6);
        nums.filter(new FilterFunction<Integer>() {
            @Override
            public boolean filter(Integer integer) throws Exception {
                return integer > 3;
            }
        }).print();//输出4,5,6


    }

    //静态内部类:import org.apache.flink.api.common.functions.MapFunction;
    static class IntegerOperations implements MapFunction<Tuple2<Integer,Integer>, Integer> {//MapFunction函数定义为输入是个元祖，输出是个Integer
        @Override
        public Integer map(Tuple2<Integer, Integer> input) throws Exception {
            return input.f0 + input.f1;//f0就是元祖中的第一个元素
        }
    }


}
